package org.lzfto.flink.demo.business.flinkTest.flinkObjects.function.process;

import java.math.BigDecimal;

import org.apache.flink.api.common.functions.OpenContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.lzfto.flink.demo.business.orders.OrderCommon;
import org.lzfto.flink.demo.business.orders.SimpleOrderAgg;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
*订单分组处理函数
*<br/> 不会进来一条就出去一条，而是
*@author lzfto
*@创建时间 2025年6月20日 09:39:28
*@param T  type of key
*/
public class  AggOrderKeyedProcessFun<T> extends KeyedProcessFunction<T, SimpleOrderAgg, SimpleOrderAgg> {
	
	private final Logger LOGGER = LoggerFactory.getLogger(getClass());
	
	private static final long serialVersionUID = 1L;
	private String grpType;
	private transient ValueState<SimpleOrderAgg> agg = null;

	
	public AggOrderKeyedProcessFun(String grpType) {
		super();
		this.grpType=grpType;
	}
	
	
	@Override
	public void open(OpenContext openContext) {
		// 初始化统计值
		TypeInformation<SimpleOrderAgg> aggType = TypeInformation.of(SimpleOrderAgg.class);
		ValueStateDescriptor<SimpleOrderAgg> aggDescriptor = new ValueStateDescriptor<>("agg", aggType);
		agg = getRuntimeContext().getState(aggDescriptor);
	}

	@Override
	public void processElement(SimpleOrderAgg value, Context ctx, Collector<SimpleOrderAgg> out)
			throws Exception {
		SimpleOrderAgg tmp = agg.value();
		if (tmp == null) {
	        tmp = new SimpleOrderAgg();	        
	        tmp.setCount(0);
	        tmp.setTotal(BigDecimal.ZERO);
	        if (grpType.equals(OrderCommon.GRP_TYPE_SEX)) {
	        	tmp.setSex(value.getSex());	
	        }
	        else if (grpType.equals(OrderCommon.GRP_TYPE_ZONE)) {
	        	tmp.setZoneName(value.getZoneName());	
	        }
	        else {
	        	tmp.setZoneName(value.getZoneName());
		        tmp.setSex(value.getSex());	
	        }
	    }
		tmp.setCount(tmp.getCount() + 1);
		tmp.setTotal(tmp.getTotal().add(value.getTotal()));
		agg.update(tmp);
		out.collect(agg.value());
	}
	
	@Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<SimpleOrderAgg> out) throws Exception {
        // 当前定时器已过期，清除状态
        agg.clear();
        LOGGER.warn("状态已清除: key = " + ctx.getCurrentKey());
    }
}
